package org.huangrui.spark.scala.streaming

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * @author hr
 * @create 2020-12-26 17:08 
 */
object SparkStreaming11_Req1 {
  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
    val sc: StreamingContext = new StreamingContext(sparkConf, Seconds(3))

    //    1.定义 Kafka 参数
    val kafkaPara: Map[String, Object] = Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "hadoop210:9092,hadoop211:9092,hadoop212:9092",
      ConsumerConfig.GROUP_ID_CONFIG -> "spark",
      "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
      "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
    )
    //    2.读取 Kafka 数据创建 DStream
    val kafkaDStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](sc,
      LocationStrategies.PreferConsistent,//采集的节点和计算的节点该如何做匹配，类似于spark core中的首选位置
      ConsumerStrategies.Subscribe[String, String](Set("spark"), kafkaPara))
    //3.将每条消息的 KV 取出
    val dstream: DStream[String] = kafkaDStream.map(record => record.value())
    //4.打印
    dstream.print()

    sc.start()
    sc.awaitTermination()
  }
}
